package com.github.common.extend.demo.config.transaction.distributed.message;

import com.github.common.extend.demo.config.transaction.TransactionSubmitEventListener;
import lombok.extern.slf4j.Slf4j;
import java.util.Iterator;
import java.util.Objects;

/**
 * 消息管理器
 * 利用 本地消息表+本地事务 实现业务和消息推送的最终一致性
 * @author X1993
 * @date 2022/2/14
 * @description
 */
@Slf4j
public class MessageManagerImpl<M extends XMessage> implements MessageManager<M> {

    private XMessageStorage<M> messageStorage;

    private XMessageSender<M> messageSender;

    private MessageManagerProperties messageManagerProperties;

    private TransactionSubmitEventListener transactionListener;

    public MessageManagerImpl(XMessageStorage<M> messageStorage,
                              XMessageSender<M> messageSender,
                              MessageManagerProperties messageManagerProperties,
                              TransactionSubmitEventListener transactionListener)
    {
        Objects.requireNonNull(messageStorage);
        Objects.requireNonNull(messageSender);
        Objects.requireNonNull(messageManagerProperties);
        Objects.requireNonNull(transactionListener);

        this.messageStorage = messageStorage;
        this.messageSender = messageSender;
        this.messageManagerProperties = messageManagerProperties;
        this.transactionListener = transactionListener;
    }

    @Override
    public void add(M message)
    {
        init(message);

        messageStorage.save(message);

        /*
        理论上不监听事务提交只通过 （定时任务异步扫描推送） 也可以实现
        但这样做有两个优点：
            1.消息推送延迟更低，不需要等到下一个定时周期
            2.定时扫描更多起到异常情况下的补偿重试作用，这样可以降低定时任务调用频率，减轻系统压力
        */
        transactionListener.register(() -> sendMessage(message));
    }

    /**
     * 消息初始化
     * @param message
     * @return
     */
    private XMessage init(M message)
    {
        return message;
    }

    /**
     * 推送消息
     * @param message
     * @return 消息是否推送成功
     */
    private void sendMessage(M message)
    {
        message.setSendCount(message.getSendCount() + 1);
        try {
            log.debug("尝试发送消息，{}" ,message);
            messageSender.send(message);
            sendSuccess(message);
        } catch (Exception e){
            log.error("消息推送异常" ,e);
            sendFail(message);
        }
    }

    /**
     * 消息发送失败流程
     * @param message
     */
    private void sendFail(M message)
    {
        log.warn("消息发送失败,{}" ,message);
        if (message.getSendCount() - 1 >= messageManagerProperties.getMaxRetryCount()){
            message.setStatus(XMessage.FAIL);
            messageStorage.update(message);
        }else {
            message.setStatus(XMessage.FAIL_RETRY);
            messageStorage.update(message);
        }
    }

    /**
     * 消息发送成功流程
     * @param message
     */
    private void sendSuccess(M message)
    {
        log.debug("消息[{}]第[{}]次发送成功" ,message.getId() ,message.getSendCount());
        message.setStatus(XMessage.SUCCESS);
        messageStorage.update(message);
    }

    /**
     * 由于每次消息记录之后都会注册一个事务提交事件用于发送消息，所以这个方法用于失败补偿即可
     * @return
     */
    public void makeUpHandler()
    {
        log.debug("执行补偿程序");
        Iterator<M> iterator = messageStorage.unsentMessages();
        while (iterator.hasNext()){
            M message = iterator.next();
            sendMessage(message);
        }
    }

}
